package com.hmeter.hadoop.mapreduce.jobtracker;

import java.io.IOException;

import org.apache.hadoop.mapred.Counters.Counter;
import org.apache.hadoop.mapred.Counters.Group;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobID;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.mapred.Counters;
import org.apache.hadoop.mapred.TaskReport;

public class TaskCountersMetric extends JobCountersMetric {

	public TaskCountersMetric() {};
	
	public String getTaskCountersMetric(String jobTracker, String jobId, String groupName) throws IOException {
		StringBuilder taskCounters = new StringBuilder();

		JobID id = convertToJobId(jobId);
		JobClient client = prepareJobClient(jobTracker);
		RunningJob job = client.getJob(id);
		TaskReport[] mapTaskReports = client.getMapTaskReports(id);
		TaskReport[] reduceTaskReports = client.getReduceTaskReports(id);

		taskCounters.append("<taskmetric jobid='"+jobId+"' jobname='"+job.getJobName()+"'>\n");
		taskCounters.append(" <mapTasks>\n");

		for (TaskReport mapTaskReport : mapTaskReports)
		{
			taskCounters.append("  <task id='"+mapTaskReport.getTaskID().toString()+"'>\n");
			taskCounters.append("   <counters>\n");

			Counters counters = mapTaskReport.getCounters();
			
			if (!groupName.equals(""))
			{
				for (Group counterGroup : counters)
				{
					if (counterGroup.getDisplayName().equals(groupName))
					{
						for (Counter counter : counterGroup)
						{	
							taskCounters.append("    <counter name='" + counter.getDisplayName() + "' value='" + counter.getValue() + "'>\n");
						}
						break;
					}	
				}
			}
			else
			{
				for (Group counterGroup : counters)
				{
					for (Counter counter : counterGroup)
					{					
						taskCounters.append("    <counter name='" + counter.getDisplayName() + "' value='" + counter.getValue() + "'>\n");
					}	
				}
			}

			taskCounters.append("   </counters>\n");
			taskCounters.append("  </task>\n");			
		}

		taskCounters.append(" </mapTasks>\n");
		taskCounters.append(" <reduceTasks>\n");

		for (TaskReport reduceTaskReport : reduceTaskReports)
		{
			taskCounters.append("  <task id='"+reduceTaskReport.getTaskID().toString()+"'>\n");
			taskCounters.append("   <counters>\n");

			Counters counters = reduceTaskReport.getCounters();

			if (!groupName.equals(""))
			{
				for (Group counterGroup : counters)
				{
					if (counterGroup.getDisplayName().equals(groupName))
					{
						for (Counter counter : counterGroup)
						{	
							taskCounters.append("    <counter name='" + counter.getDisplayName() + "' value='" + counter.getValue() + "'>\n");
						}
						break;
					}	
				}
			}
			else
			{
				for (Group counterGroup : counters)
				{
					for (Counter counter : counterGroup)
					{					
						taskCounters.append("    <counter name='" + counter.getDisplayName() + "' value='" + counter.getValue() + "'>\n");
					}	
				}
			}
			
			taskCounters.append("   </counters>\n");
			taskCounters.append("  </task>\n");			
		}

		taskCounters.append(" </reduceTasks>\n");
		taskCounters.append("</taskmetric>");

		return taskCounters.toString();
	}

	public String getTaskStatusMetric(String jobTracker, String jobId) throws IOException {
		StringBuilder taskStatistics = new StringBuilder();
		long taskDuration = 0;
		String duration = "";

		JobID id = convertToJobId(jobId);
		JobClient client = prepareJobClient(jobTracker);
		RunningJob job = client.getJob(id);

		TaskReport[] mapTaskReports = client.getMapTaskReports(id);
		TaskReport[] reduceTaskReports = client.getReduceTaskReports(id);

		taskStatistics.append("<taskmetric jobid='"+jobId+"' jobname='"+job.getJobName()+"'>\n");
		taskStatistics.append(" <mapTasks>\n");

		for (TaskReport mapTaskReport : mapTaskReports)
		{
			taskDuration = mapTaskReport.getFinishTime() - mapTaskReport.getStartTime();

			if (taskDuration < 0)
			{
				duration = "N/A";
			}
			else
			{
				duration = String.valueOf(taskDuration);
			}

			double progress = mapTaskReport.getProgress() * 100;
			String taskProgress = Double.toString(progress) + "%";

			taskStatistics.append("  <task id='"+mapTaskReport.getTaskID().toString()+"'>\n");
			taskStatistics.append("   <progress>"+taskProgress+"</progress>\n");
			taskStatistics.append("   <duration>"+duration+"</duration>\n");
			taskStatistics.append("   <status>"+mapTaskReport.getCurrentStatus().toString()+"</status>\n");
			taskStatistics.append("  </task>\n");	
		}

		taskStatistics.append(" </mapTasks>\n");

		taskStatistics.append(" <reduceTasks>\n");

		for (TaskReport reduceTaskReport : reduceTaskReports)
		{
			taskDuration = reduceTaskReport.getFinishTime() - reduceTaskReport.getStartTime();

			if (taskDuration < 0)
			{
				duration = "N/A";
			}
			else
			{
				duration = String.valueOf(taskDuration);
			}

			double progress = reduceTaskReport.getProgress() * 100;
			String taskProgress = Double.toString(progress) + "%";

			taskStatistics.append("  <task id='"+reduceTaskReport.getTaskID().toString()+"'>\n");
			taskStatistics.append("   <progress>"+taskProgress+"</progress>\n");
			taskStatistics.append("   <duration>"+duration+"</duration>\n");
			taskStatistics.append("   <status>"+reduceTaskReport.getCurrentStatus().toString()+"</status>\n");
			taskStatistics.append("  </task>\n");					
		}

		taskStatistics.append(" </reduceTasks>\n");
		taskStatistics.append("</taskmetric>");

		return taskStatistics.toString();
	}
}
